package com.niit.MostPopularMovieType;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

public class MPMTypeReducer extends Reducer<Text, MPMTypeBean, Text, MPMTypeBean> {
    // 定义一个TreeMap作为存储数据的容器（按key排序）
    private TreeMap<MPMTypeBean, Text> flowMap = new TreeMap<>();

    @Override
    protected void reduce(Text key, Iterable<MPMTypeBean> values, Context context) throws IOException, InterruptedException {
        for (MPMTypeBean value :
                values) {
            MPMTypeBean flowBean = new MPMTypeBean();
            flowBean.set(value.getRate(), value.getRank(), value.getTypes(), value.getRegions(), value.getTitle(),
                    value.getRelease_data(), value.getActor_count(), value.getVote_count(), value.getScore(), value.getActors());

            Text phone = new Text(key);

//             1 向treeMap集合中添加数据
            flowMap.put(flowBean, phone);

            // 2 限制TreeMap数据量，超过10条就删除掉流量最小的一条数据
            if (flowMap.size() > 10) {
                flowMap.remove(flowMap.lastKey());
            }
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        // 3 遍历集合，输出数据
        Iterator<MPMTypeBean> iterator = flowMap.keySet().iterator();

        while (iterator.hasNext()) {
            MPMTypeBean valueOut = iterator.next();
            Text keyOut = flowMap.get(valueOut);

            context.write(keyOut, valueOut);
        }

    }
}
